package com.yamed.bus.service.kakfa;

import com.alibaba.fastjson.JSONObject;
import com.yamed.bus.service.EmailService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * @Author: wchen
 * @Description:
 * @Date: Created in 2019-08-13 10:28
 * Copyright 本内容仅限于杭州阿拉丁信息科技股份有限公司内部传阅，禁止外泄以及用于其他的商业目的
 */
@Component
public class KafKaConsumer {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private EmailService emailService;

    @KafkaListener(topics = "${kafka.topic-email}", groupId = "${kafka.group-email}")
    public void listen(ConsumerRecord<?, ?> record) throws Exception {
        logger.info("topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
        logger.info("KafKaConsumer content" + record.value().toString());
        JSONObject requestData = JSONObject.parseObject(record.value().toString());
        String content = requestData.getString("content");
        String title = requestData.getString("title");
        String toAddress = requestData.getString("toAddress");
        emailService.sendSimpleMail(toAddress, title, content);
    }


    @KafkaListener(containerFactory = "batchFactory", topics = "${kafka.topic-my-topic}", groupId = "${kafka.group-my-topic}")
    public void test(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment) throws Exception {
        logger.info("*******************" + records.size() + "*********************");
//        Thread.sleep(6000);
        logger.info("KafKaConsumer test " + JSONObject.toJSONString(records, true));
        records.forEach(record -> {
            logger.info(String.format("topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value()));
            logger.info("KafKaConsumer content" + record.value().toString());
        });
        acknowledgment.acknowledge();
    }
}
